--------------------------sql任务说明----------------------------------
-- @author :liuxiaojie05
-- @descrbe :投递阅读数据 ods到dwd udtf 函数
-- 具体可见：http://wbook.58corp.com/books/jianlibook/data/wmb/16503/
-- @udf jar包 : bigdata_realtime_flink_sql_udf-1.0-testlxj-jar-with-dependencies.jar
-- 下游使用方：olap、dmp
--------------------------sql任务说明----------------------------------

-- 加载函数
CREATE FUNCTION ZpDWDDeliveryReadUdtf as 'udtf.etl.dwd.ZpDWDDeliveryReadUdtf';

-- source
CREATE TABLE ods_zp_delivery_read (
	log  STRING
) with (
	'connector' = 'wmb',
	'topic' = '16503',
	'client.id' = '6',
	'properties.key' = 'F5993419EE58377033B8B7792814A71F',
	'properties.ip' = 'esb.registryserver.58dns.org',
	'properties.port' = '21010',
	'format' = 'raw');


-- sink
create table dwd_zp_delivery_read
(
 		ruserid      string,
        euserid      string,
        infoid       string,
        resumeid     string,
        addtime      string,
        content      string,
        messagetype  string,
        wmb_time  BIGINT,
        sink_time string,
        e_time    BIGINT
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_dwd_zp_delivery_read',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin_dwd_zp_delivery_read-sql-prod',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_delivery_read-Cqn3D',
'format' = 'json'
);

-- sql-etl
insert into dwd_zp_delivery_read
select pp.* from ods_zp_delivery_read, LATERAL TABLE(ZpDWDDeliveryReadUdtf(log)) as pp;